package com.datagateway.component;

import com.datagateway.model.ProcessedData;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 压力测试运行器
 * 负责执行系统压力测试和性能基准测试
 * 
 * @author Data Gateway Team
 * @version 1.0.0
 */
@Component
public class StressTestRunner {

    private static final Logger logger = LoggerFactory.getLogger(StressTestRunner.class);

    @Autowired
    private DataBuffer dataBuffer;

    @Autowired
    private PerformanceOptimizer performanceOptimizer;

    @Autowired
    private AlertManager alertManager;

    /**
     * 压力测试统计
     */
    private final AtomicLong totalTestMessages = new AtomicLong(0);
    private final AtomicLong successfulTests = new AtomicLong(0);
    private final AtomicLong failedTests = new AtomicLong(0);

    /**
     * 执行压力测试
     * 
     * @param testConfig 测试配置
     * @return 测试结果
     */
    public StressTestResult runStressTest(StressTestConfig testConfig) {
        logger.info("开始执行压力测试: {}", testConfig);
        
        long startTime = System.currentTimeMillis();
        
        try {
            // 创建线程池
            ExecutorService executor = Executors.newFixedThreadPool(testConfig.getThreadCount());
            List<Future<TestThreadResult>> futures = new ArrayList<>();
            
            // 启动测试线程
            for (int i = 0; i < testConfig.getThreadCount(); i++) {
                Future<TestThreadResult> future = executor.submit(
                    new StressTestThread(i, testConfig, dataBuffer)
                );
                futures.add(future);
            }
            
            // 等待所有线程完成
            List<TestThreadResult> results = new ArrayList<>();
            for (Future<TestThreadResult> future : futures) {
                try {
                    results.add(future.get(testConfig.getTimeoutSeconds(), TimeUnit.SECONDS));
                } catch (TimeoutException e) {
                    logger.warn("测试线程超时", e);
                    future.cancel(true);
                }
            }
            
            // 关闭线程池
            executor.shutdown();
            
            long endTime = System.currentTimeMillis();
            long totalTime = endTime - startTime;
            
            // 计算测试结果
            StressTestResult testResult = calculateTestResult(results, totalTime, testConfig);
            
            logger.info("压力测试完成: 总时间={} ms, 成功率={:.2f}%", 
                       totalTime, testResult.getSuccessRate());
            
            return testResult;
            
        } catch (Exception e) {
            logger.error("压力测试执行失败", e);
            alertManager.sendSystemErrorAlert("压力测试执行失败", e.getMessage());
            throw new RuntimeException("压力测试执行失败", e);
        }
    }

    /**
     * 执行性能基准测试
     * 
     * @param benchmarkConfig 基准测试配置
     * @return 基准测试结果
     */
    public BenchmarkResult runBenchmarkTest(BenchmarkConfig benchmarkConfig) {
        logger.info("开始执行性能基准测试: {}", benchmarkConfig);
        
        long startTime = System.currentTimeMillis();
        
        try {
            // 获取测试前的性能指标
            PerformanceOptimizer.PerformanceMetrics beforeMetrics = performanceOptimizer.getCurrentPerformanceMetrics();
            
            // 执行基准测试
            StressTestResult testResult = runStressTest(benchmarkConfig.toStressTestConfig());
            
            // 获取测试后的性能指标
            PerformanceOptimizer.PerformanceMetrics afterMetrics = performanceOptimizer.getCurrentPerformanceMetrics();
            
            long endTime = System.currentTimeMillis();
            long totalTime = endTime - startTime;
            
            // 计算基准测试结果
            BenchmarkResult benchmarkResult = new BenchmarkResult(
                testResult,
                beforeMetrics,
                afterMetrics,
                totalTime,
                benchmarkConfig
            );
            
            logger.info("性能基准测试完成: 吞吐量={:.2f} msg/s, 平均延迟={:.2f} ms", 
                       benchmarkResult.getThroughput(), benchmarkResult.getAverageLatency());
            
            return benchmarkResult;
            
        } catch (Exception e) {
            logger.error("性能基准测试执行失败", e);
            throw new RuntimeException("性能基准测试执行失败", e);
        }
    }

    /**
     * 计算测试结果
     * 
     * @param results 线程测试结果列表
     * @param totalTime 总测试时间
     * @param testConfig 测试配置
     * @return 压力测试结果
     */
    private StressTestResult calculateTestResult(List<TestThreadResult> results, long totalTime, StressTestConfig testConfig) {
        long totalMessages = 0;
        long successfulMessages = 0;
        long failedMessages = 0;
        long totalLatency = 0;
        long minLatency = Long.MAX_VALUE;
        long maxLatency = 0;
        
        for (TestThreadResult result : results) {
            totalMessages += result.getTotalMessages();
            successfulMessages += result.getSuccessfulMessages();
            failedMessages += result.getFailedMessages();
            totalLatency += result.getTotalLatency();
            minLatency = Math.min(minLatency, result.getMinLatency());
            maxLatency = Math.max(maxLatency, result.getMaxLatency());
        }
        
        double successRate = totalMessages > 0 ? (double) successfulMessages / totalMessages * 100 : 0;
        double throughput = totalTime > 0 ? (double) totalMessages / totalTime * 1000 : 0; // 消息/秒
        double averageLatency = successfulMessages > 0 ? (double) totalLatency / successfulMessages : 0;
        
        return new StressTestResult(
            testConfig,
            totalMessages,
            successfulMessages,
            failedMessages,
            successRate,
            throughput,
            averageLatency,
            minLatency == Long.MAX_VALUE ? 0 : minLatency,
            maxLatency,
            totalTime,
            System.currentTimeMillis()
        );
    }

    /**
     * 压力测试线程
     */
    private static class StressTestThread implements Callable<TestThreadResult> {
        private final int threadId;
        private final StressTestConfig testConfig;
        private final DataBuffer dataBuffer;
        
        private final AtomicLong totalMessages = new AtomicLong(0);
        private final AtomicLong successfulMessages = new AtomicLong(0);
        private final AtomicLong failedMessages = new AtomicLong(0);
        private final AtomicLong totalLatency = new AtomicLong(0);
        private volatile long minLatency = Long.MAX_VALUE;
        private volatile long maxLatency = 0;

        public StressTestThread(int threadId, StressTestConfig testConfig, DataBuffer dataBuffer) {
            this.threadId = threadId;
            this.testConfig = testConfig;
            this.dataBuffer = dataBuffer;
        }

        @Override
        public TestThreadResult call() throws Exception {
            long startTime = System.currentTimeMillis();
            long endTime = startTime + testConfig.getDurationSeconds() * 1000;
            
            while (System.currentTimeMillis() < endTime && totalMessages.get() < testConfig.getMaxMessagesPerThread()) {
                try {
                    // 创建测试数据
                    ProcessedData testData = createTestData(threadId, totalMessages.get());
                    
                    // 记录开始时间
                    long messageStartTime = System.currentTimeMillis();
                    
                    // 发送数据
                    dataBuffer.addData(testData);
                    
                    // 记录延迟
                    long latency = System.currentTimeMillis() - messageStartTime;
                    totalLatency.addAndGet(latency);
                    minLatency = Math.min(minLatency, latency);
                    maxLatency = Math.max(maxLatency, latency);
                    
                    // 更新统计
                    totalMessages.incrementAndGet();
                    successfulMessages.incrementAndGet();
                    
                    // 控制发送频率
                    if (testConfig.getMessagesPerSecond() > 0) {
                        long sleepTime = 1000 / testConfig.getMessagesPerSecond();
                        if (sleepTime > 0) {
                            Thread.sleep(sleepTime);
                        }
                    }
                    
                } catch (Exception e) {
                    failedMessages.incrementAndGet();
                    logger.warn("压力测试线程 {} 发送消息失败", threadId, e);
                }
            }
            
            return new TestThreadResult(
                threadId,
                totalMessages.get(),
                successfulMessages.get(),
                failedMessages.get(),
                totalLatency.get(),
                minLatency == Long.MAX_VALUE ? 0 : minLatency,
                maxLatency
            );
        }

        /**
         * 创建测试数据
         */
        private ProcessedData createTestData(int threadId, long messageId) {
            ProcessedData data = new ProcessedData();
            data.setId("stress-test-" + threadId + "-" + messageId);
            data.setOriginalData(String.format(
                "{\"threadId\": %d, \"messageId\": %d, \"timestamp\": %d, \"data\": \"stress test data\"}",
                threadId, messageId, System.currentTimeMillis()
            ));
            
            // 设置转换后的数据
            Map<String, Object> transformedData = new java.util.HashMap<>();
            transformedData.put("threadId", threadId);
            transformedData.put("messageId", messageId);
            transformedData.put("timestamp", System.currentTimeMillis());
            transformedData.put("data", "stress test data");
            transformedData.put("testType", "stress_test");
            data.setTransformedData(transformedData);
            
            data.setSourceTopic("stress-test-topic");
            data.setPartition(threadId % 10);
            data.setOffset(messageId);
            data.setStatus(ProcessedData.ProcessStatus.SUCCESS);
            data.setDataSize(data.getOriginalData().getBytes().length);
            
            return data;
        }
    }

    /**
     * 压力测试配置类
     */
    public static class StressTestConfig {
        private final int threadCount;
        private final int durationSeconds;
        private final int maxMessagesPerThread;
        private final int messagesPerSecond;
        private final int timeoutSeconds;

        public StressTestConfig(int threadCount, int durationSeconds, int maxMessagesPerThread, 
                              int messagesPerSecond, int timeoutSeconds) {
            this.threadCount = threadCount;
            this.durationSeconds = durationSeconds;
            this.maxMessagesPerThread = maxMessagesPerThread;
            this.messagesPerSecond = messagesPerSecond;
            this.timeoutSeconds = timeoutSeconds;
        }

        // Getter方法
        public int getThreadCount() { return threadCount; }
        public int getDurationSeconds() { return durationSeconds; }
        public int getMaxMessagesPerThread() { return maxMessagesPerThread; }
        public int getMessagesPerSecond() { return messagesPerSecond; }
        public int getTimeoutSeconds() { return timeoutSeconds; }

        @Override
        public String toString() {
            return String.format("StressTestConfig{threadCount=%d, durationSeconds=%d, " +
                               "maxMessagesPerThread=%d, messagesPerSecond=%d, timeoutSeconds=%d}",
                               threadCount, durationSeconds, maxMessagesPerThread, messagesPerSecond, timeoutSeconds);
        }
    }

    /**
     * 基准测试配置类
     */
    public static class BenchmarkConfig {
        private final int threadCount;
        private final int durationSeconds;
        private final int maxMessagesPerThread;
        private final int messagesPerSecond;
        private final int timeoutSeconds;
        private final String benchmarkName;

        public BenchmarkConfig(String benchmarkName, int threadCount, int durationSeconds, 
                             int maxMessagesPerThread, int messagesPerSecond, int timeoutSeconds) {
            this.benchmarkName = benchmarkName;
            this.threadCount = threadCount;
            this.durationSeconds = durationSeconds;
            this.maxMessagesPerThread = maxMessagesPerThread;
            this.messagesPerSecond = messagesPerSecond;
            this.timeoutSeconds = timeoutSeconds;
        }

        // Getter方法
        public String getBenchmarkName() { return benchmarkName; }
        public int getThreadCount() { return threadCount; }
        public int getDurationSeconds() { return durationSeconds; }
        public int getMaxMessagesPerThread() { return maxMessagesPerThread; }
        public int getMessagesPerSecond() { return messagesPerSecond; }
        public int getTimeoutSeconds() { return timeoutSeconds; }

        public StressTestConfig toStressTestConfig() {
            return new StressTestConfig(threadCount, durationSeconds, maxMessagesPerThread, 
                                      messagesPerSecond, timeoutSeconds);
        }

        @Override
        public String toString() {
            return String.format("BenchmarkConfig{name='%s', threadCount=%d, durationSeconds=%d, " +
                               "maxMessagesPerThread=%d, messagesPerSecond=%d, timeoutSeconds=%d}",
                               benchmarkName, threadCount, durationSeconds, maxMessagesPerThread, 
                               messagesPerSecond, timeoutSeconds);
        }
    }

    /**
     * 线程测试结果类
     */
    public static class TestThreadResult {
        private final int threadId;
        private final long totalMessages;
        private final long successfulMessages;
        private final long failedMessages;
        private final long totalLatency;
        private final long minLatency;
        private final long maxLatency;

        public TestThreadResult(int threadId, long totalMessages, long successfulMessages, 
                              long failedMessages, long totalLatency, long minLatency, long maxLatency) {
            this.threadId = threadId;
            this.totalMessages = totalMessages;
            this.successfulMessages = successfulMessages;
            this.failedMessages = failedMessages;
            this.totalLatency = totalLatency;
            this.minLatency = minLatency;
            this.maxLatency = maxLatency;
        }

        // Getter方法
        public int getThreadId() { return threadId; }
        public long getTotalMessages() { return totalMessages; }
        public long getSuccessfulMessages() { return successfulMessages; }
        public long getFailedMessages() { return failedMessages; }
        public long getTotalLatency() { return totalLatency; }
        public long getMinLatency() { return minLatency; }
        public long getMaxLatency() { return maxLatency; }
    }

    /**
     * 压力测试结果类
     */
    public static class StressTestResult {
        private final StressTestConfig testConfig;
        private final long totalMessages;
        private final long successfulMessages;
        private final long failedMessages;
        private final double successRate;
        private final double throughput;
        private final double averageLatency;
        private final long minLatency;
        private final long maxLatency;
        private final long totalTime;
        private final long timestamp;

        public StressTestResult(StressTestConfig testConfig, long totalMessages, long successfulMessages, 
                              long failedMessages, double successRate, double throughput, double averageLatency,
                              long minLatency, long maxLatency, long totalTime, long timestamp) {
            this.testConfig = testConfig;
            this.totalMessages = totalMessages;
            this.successfulMessages = successfulMessages;
            this.failedMessages = failedMessages;
            this.successRate = successRate;
            this.throughput = throughput;
            this.averageLatency = averageLatency;
            this.minLatency = minLatency;
            this.maxLatency = maxLatency;
            this.totalTime = totalTime;
            this.timestamp = timestamp;
        }

        // Getter方法
        public StressTestConfig getTestConfig() { return testConfig; }
        public long getTotalMessages() { return totalMessages; }
        public long getSuccessfulMessages() { return successfulMessages; }
        public long getFailedMessages() { return failedMessages; }
        public double getSuccessRate() { return successRate; }
        public double getThroughput() { return throughput; }
        public double getAverageLatency() { return averageLatency; }
        public long getMinLatency() { return minLatency; }
        public long getMaxLatency() { return maxLatency; }
        public long getTotalTime() { return totalTime; }
        public long getTimestamp() { return timestamp; }
    }

    /**
     * 基准测试结果类
     */
    public static class BenchmarkResult {
        private final StressTestResult stressTestResult;
        private final PerformanceOptimizer.PerformanceMetrics beforeMetrics;
        private final PerformanceOptimizer.PerformanceMetrics afterMetrics;
        private final long totalTime;
        private final BenchmarkConfig benchmarkConfig;

        public BenchmarkResult(StressTestResult stressTestResult, 
                             PerformanceOptimizer.PerformanceMetrics beforeMetrics,
                             PerformanceOptimizer.PerformanceMetrics afterMetrics,
                             long totalTime, BenchmarkConfig benchmarkConfig) {
            this.stressTestResult = stressTestResult;
            this.beforeMetrics = beforeMetrics;
            this.afterMetrics = afterMetrics;
            this.totalTime = totalTime;
            this.benchmarkConfig = benchmarkConfig;
        }

        // Getter方法
        public StressTestResult getStressTestResult() { return stressTestResult; }
        public PerformanceOptimizer.PerformanceMetrics getBeforeMetrics() { return beforeMetrics; }
        public PerformanceOptimizer.PerformanceMetrics getAfterMetrics() { return afterMetrics; }
        public long getTotalTime() { return totalTime; }
        public BenchmarkConfig getBenchmarkConfig() { return benchmarkConfig; }
        
        public double getThroughput() { return stressTestResult.getThroughput(); }
        public double getAverageLatency() { return stressTestResult.getAverageLatency(); }
        public double getSuccessRate() { return stressTestResult.getSuccessRate(); }
        
        public double getMemoryIncrease() {
            return afterMetrics.getUsedHeapMemory() - beforeMetrics.getUsedHeapMemory();
        }
        
        public double getGcTimeIncrease() {
            return afterMetrics.getGcTime() - beforeMetrics.getGcTime();
        }
    }
}
